package com.etl.component.basic;

import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.excepiton.ComponentException;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.MergeColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 描述：
 * 汇聚合并
 * @author xianggj
 * @Date 2021/10/12 18:10
 **/
public class DataCollisionComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        if (datas.size() != 2){
            throw new ComponentException("汇聚合并输入端必须为两个");
        }
        Map<String, PortData> nodeMap = datas.stream().collect(Collectors.toMap(PortData::getNodeId, Function.identity()));
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<MergeColumnDTO> mergeColumnDTOS =
                (List<MergeColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        String mergeType = (String) paramMap.getOrDefault("mergeType", "join");
        String mergeWay = (String) paramMap.getOrDefault("mergeWay", "left");
        //这里是副表
        String tableId = (String) paramMap.get(ComponentConstants.TABLE_NAME);
        if (StrUtil.isBlank(tableId) || !nodeMap.containsKey(tableId)){
            throw new ComponentException("需要合并的表未找到，id为"+ tableId);
        }
        PortData[] setDatas = comfireMaster(nodeMap, tableId);
        SetData input0 = (SetData)setDatas[0].getValue();
        SetData input1 = (SetData)setDatas[1].getValue();
        DataSet<Row> dataset0 = (DataSet<Row>) input0.getDataset();
        List<TableColumn> columns0 = input0.getColumns();
        DataSet<Row> dataset1 = (DataSet<Row>) input1.getDataset();
        List<TableColumn> columns1 = input1.getColumns();
        //设置合并列坐标
        setIndex(mergeColumnDTOS, columns0, columns1, mergeType);
        //记录结果列组装哪些数据
        List<int[]> colIndex = new ArrayList<>();
        List<TableColumn> columns = getNewTableColumns(nodeMap, mergeColumnDTOS,
                mergeType, mergeWay, tableId, colIndex);
        RowTypeInfo typeInfo = changeType(columns);
        DataSet<Row> result = null;
        if(MergeColumnDTO.JOIN.equalsIgnoreCase(mergeType)){
            if (colIndex.size() != columns.size()){
                throw new ComponentException("代码错误");
            }
            result = getRowDataSet(mergeWay, dataset0, dataset1, colIndex, mergeColumnDTOS);
            result = new MapOperator<Row, Row>(result, typeInfo, e -> e,  Utils.getCallLocationName());
        } else {
            dataset0 = new MapOperator<>(dataset0, typeInfo, e -> e,  Utils.getCallLocationName());
            dataset1 = covertDataset1(mergeColumnDTOS, dataset1, typeInfo);
            result = dataset0.union(dataset1);
        }
        return execute(result, columns, "执行汇聚合并操作");
    }

    /**
     * 转换dataset1
     * @param mergeColumnDTOS
     * @param dataset1
     * @param typeInfo
     * @return
     */
    private DataSet<Row> covertDataset1(List<MergeColumnDTO> mergeColumnDTOS, DataSet<Row> dataset1, RowTypeInfo typeInfo) {
        List<Integer> indexList = mergeColumnDTOS.stream()
                .map(e -> e.getSlaveIndex()).collect(Collectors.toList());
        dataset1 = new MapOperator<>(dataset1, typeInfo, e -> {
            Object[] objs = new Object[indexList.size()];
            int i = 0 ;
            for (Integer slaveIndex : indexList) {
                if (slaveIndex == -1){
                    objs[i] = null;
                } else {
                    objs[i] = e.getField(slaveIndex);
                }
                i++;
            }
            return Row.of(objs);
        },  Utils.getCallLocationName());
        return dataset1;
    }

    /**
     * 组合数据
     * @param mergeWay 合并方式
     * @param dataset0 数据集1
     * @param dataset1 数据集2
     * @param colIndex 列位置
     * @param mergeColumnDTOS
     * @return
     */
    private DataSet<Row> getRowDataSet(String mergeWay, DataSet<Row> dataset0,
                                       DataSet<Row> dataset1, List<int[]> colIndex,
                                       List<MergeColumnDTO> mergeColumnDTOS) {
        int[] wheres = new int[mergeColumnDTOS.size()];
        int[] equals = new int[mergeColumnDTOS.size()];
        int i = 0;
        for (MergeColumnDTO mergeColumnDTO : mergeColumnDTOS){
            wheres[i] = mergeColumnDTO.getMasterIndex();
            equals[i] = mergeColumnDTO.getSlaveIndex();
            i++;
        }
        switch (mergeWay){
            case MergeColumnDTO.LEFT:
                return dataset0.leftOuterJoin(dataset1)
                        .where(wheres).equalTo(equals)
                        .with(getLeftJoinFunction(colIndex));
            case MergeColumnDTO.RIGHT:
                return dataset0.rightOuterJoin(dataset1)
                        .where(wheres).equalTo(equals)
                        .with(getRightJoinFunction(colIndex));
            case MergeColumnDTO.AND_SET:
                return dataset0.fullOuterJoin(dataset1)
                        .where(wheres).equalTo(equals)
                        .with(getAllJoinFunction(colIndex));
            case MergeColumnDTO.INTERSECTION:
                return dataset0.join(dataset1)
                        .where(wheres).equalTo(equals)
                        .with(getJoinFunction(colIndex));
            default:
                throw new ComponentException("不支持当前合并方式"+ mergeWay);
        }
    }

    /**
     * 内链接（交集）
     * @param colIndex
     * @return
     */
    private JoinFunction<Row, Row, Row> getJoinFunction(List<int[]> colIndex) {
        return (a, b) -> {
            Object[] objs = new Object[colIndex.size()];
            for (int j = 0; j < colIndex.size(); j++) {
                int[] ints = colIndex.get(j);
                if (ints[0] == 0) {
                    objs[j] = a.getField(ints[1]);
                } else if (ints[0] == 1) {
                    objs[j] = b.getField(ints[1]);
                } else {
                    if (a == null) {
                        objs[j] = b.getField(ints[2]);
                    } else {
                        objs[j] = a.getField(ints[1]);
                    }
                }
            }
            return Row.of(objs);
        };
    }

    /**
     * 全链接 并集
     * @param colIndex
     * @return
     */
    private JoinFunction<Row, Row, Row> getAllJoinFunction(List<int[]> colIndex) {
        return (a,b)->{
            Object[] objs = new Object[colIndex.size()];
            for (int j = 0; j < colIndex.size(); j++) {
                int[] ints = colIndex.get(j);
                if (ints[0] == 0){
                    if (a == null){
                        objs[j] = null;
                    }else {
                        objs[j] = a.getField(ints[1]);
                    }
                } else if (ints[0] == 1) {
                    if (b == null){
                        objs[j] = null;
                    }else {
                        objs[j] = b.getField(ints[1]);
                    }
                } else {
                    if (b == null) {
                        objs[j] = a.getField(ints[1]);
                    } else {
                        objs[j] = b.getField(ints[2]);
                    }
                }
            }
            return Row.of(objs);
        };
    }

    /**
     * 右链接
     * @param colIndex
     * @return
     */
    private JoinFunction<Row, Row, Row> getRightJoinFunction(List<int[]> colIndex) {
        return (a,b)->{
            Object[] objs = new Object[colIndex.size()];
            for (int j = 0; j < colIndex.size(); j++) {
                int[] ints = colIndex.get(j);
                if (ints[0] == 0){
                    if (a == null){  //要判断如果右边没有匹配的，显示什么
                        objs[j] = null;
                    }else {
                        objs[j] = a.getField(ints[1]);
                    }
                } else if (ints[0] == 1) {
                    objs[j] = b.getField(ints[1]);
                } else {
                    if (b == null) {
                        objs[j] = a.getField(ints[1]);
                    } else {
                        objs[j] = b.getField(ints[2]);
                    }
                }
            }
            return Row.of(objs);
        };
    }

    /**
     * 左链接
     * @param colIndex
     * @return
     */
    private JoinFunction<Row, Row, Row> getLeftJoinFunction(List<int[]> colIndex) {
        return (a,b)->{
            Object[] objs = new Object[colIndex.size()];
            for (int j = 0; j < colIndex.size(); j++) {
                int[] ints = colIndex.get(j);
                if (ints[0] == 0){
                    objs[j] = a.getField(ints[1]);
                } else if  (ints[0] == 1) {
                    if (b == null){
                        objs[j] = null;
                    }else{
                        objs[j] = b.getField(ints[1]);
                    }
                } else {
                    if (b == null){
                        objs[j] = a.getField(ints[1]);
                    }else{
                        objs[j] = b.getField(ints[2]);
                    }
                }
            }
            return Row.of(objs);
        };
    }

    /**
     * 转换副表
     * 这里是离线转
     * @param dataset1
     * @param typeInfo 列类型
     * @param mergeColumnDTOS
     * @return
     */
    @Deprecated
    private DataSet<Row> covertDataset(DataSet<Row> dataset1,  RowTypeInfo typeInfo, List<MergeColumnDTO> mergeColumnDTOS) {
        ExecutionEnvironment env = dataset1.getExecutionEnvironment();
        List<Row> newCollect = new ArrayList<>();
        try {
            List<Row> collect = dataset1.collect();
            for (Row row : collect) {
                Object[] objs = new Object[mergeColumnDTOS.size()];
                int i = 0 ;
                for (MergeColumnDTO mergeColumnDTO : mergeColumnDTOS) {
                    int slaveIndex = mergeColumnDTO.getSlaveIndex();
                    if (slaveIndex == -1){
                        objs[i] = null;
                    } else {
                        objs[i] = row.getField(slaveIndex);
                    }
                    i++;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new ComponentException("副表类型转换异常", e);
        }
       return env.fromCollection(newCollect, typeInfo);
    }


    /**
     * 类型转换
     * @param columns
     * @return
     */
    protected RowTypeInfo changeType(List<TableColumn> columns) {
        String[] fieldNames = new String[columns.size()];
        TypeInformation<?>[] fieldTypes = new TypeInformation[columns.size()];
        for (int i = 0; i < columns.size() ; i++) {
            TableColumn tableColumn = columns.get(i);
            fieldTypes[i] = typeToBasicTypeInfo(tableColumn.getDataType());
            fieldNames[i] = tableColumn.getColumnName();
        }
        return new RowTypeInfo(fieldTypes, fieldNames);
    }

    /**
     * 确认主表副表
     * @param nodeMap
     * @param tableId
     * @return 主表在前 附表在后
     */
    private PortData[] comfireMaster(Map<String, PortData> nodeMap, String tableId) {
        PortData[] sets = new PortData[2];
        for (Map.Entry<String, PortData> entry:nodeMap.entrySet()) {
            if (entry.getKey().equalsIgnoreCase(tableId)){
                sets[1] = entry.getValue();
            } else {
                sets[0] = entry.getValue();
            }
        }
        return sets;
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        if (datas.size() != 2){
            throw new ComponentException("汇聚合并输入端必须为两个");
        }
        Map<String, PortData> nodeMap = datas.stream().collect(Collectors.toMap(PortData::getNodeId, Function.identity()));
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<MergeColumnDTO> mergeColumnDTOS =
                (List<MergeColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        String mergeType = (String) paramMap.getOrDefault("mergeType", "join");
        String mergeWay = (String) paramMap.getOrDefault("mergeWay", "left");
        //这里是副表
        String tableId = (String) paramMap.get(ComponentConstants.TABLE_NAME);
        if (StrUtil.isBlank(tableId) || !nodeMap.containsKey(tableId)){
            throw new ComponentException("需要合并的表未找到，id为"+ tableId);
        }
        List<int[]> colIndex = new ArrayList<>();
        List<TableColumn> columns = getNewTableColumns(nodeMap, mergeColumnDTOS, mergeType, mergeWay, tableId, colIndex);

        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    /**
     * 新解析列
     * @param nodeMap
     * @param mergeColumnDTOS
     * @param mergeType
     * @param mergeWay
     * @param tableId
     * @return
     */
    private List<TableColumn> getNewTableColumns(Map<String, PortData> nodeMap, List<MergeColumnDTO> mergeColumnDTOS,
                                                 String mergeType, String mergeWay,String tableId, List<int[]> colIndex) {
        List<TableColumn> columns = new ArrayList<>();
        PortData[] setDatas = comfireMaster(nodeMap, tableId);
        ColumnData input0 = (ColumnData)setDatas[0].getValue();
        ColumnData input1 = (ColumnData)setDatas[1].getValue();
        List<TableColumn> columns0 = input0.getColumns();
        List<TableColumn> columns1 = input1.getColumns();
        //左右合并
        Map<String, TableColumn> masterMap = columns0.stream().collect(Collectors.toMap(TableColumn::getColumnName, Function.identity()));
        Map<String, TableColumn> slaveMap = columns1.stream().collect(Collectors.toMap(TableColumn::getColumnName, Function.identity()));
        if (MergeColumnDTO.JOIN.equalsIgnoreCase(mergeType)){
            Map<String, TableColumn> mMap = new HashMap<>();
            Set<String> slaveSet = new HashSet<>();
            for (MergeColumnDTO mergeColumnDTO : mergeColumnDTOS) {
                String master = mergeColumnDTO.getMaster();
                String slave = mergeColumnDTO.getSlave();
                String result = mergeColumnDTO.getResult();
                if (masterMap.containsKey(master) && slaveMap.containsKey(slave)){
                    TableColumn masterCol = masterMap.get(master);
                    TableColumn slaveCol = slaveMap.get(slave);
                    if (typeToBasicTypeInfo(masterCol.getDataType()).equals(typeToBasicTypeInfo(slaveCol.getDataType()))){
                        TableColumn clone = masterCol.clone();
                        clone.setColumnName(result);
                        if (mMap.containsKey(master)) {
                            throw new ComponentException(String.format("主表列%s不能重复关联两次", master));
                        }
                        mMap.put(master, clone);
                        slaveSet.add(slave);
                        colIndex.add(new int[]{2, mergeColumnDTO.getMasterIndex(), mergeColumnDTO.getSlaveIndex()});
                    } else {
                        throw new ComponentException(String.format("主表列%s与副表列%s类型不匹配", master,slave));
                    }
                } else if (!masterMap.containsKey(master)){
                    throw new ComponentException(String.format("主表列%s不存在", master));
                } else {
                    throw new ComponentException(String.format("副表列%s不存在", slave));
                }

            }
            for (int i = 0; i < columns0.size(); i++) {
                TableColumn tableColumn = columns0.get(i);
                if (mMap.containsKey(tableColumn.getColumnName())){
                    columns.add(mMap.get(tableColumn.getColumnName()));
                    continue;
                }
                colIndex.add(new int[]{0,i});
                TableColumn clone = tableColumn.clone();
                columns.add(clone);
            }
            for (int i = 0; i < columns1.size(); i++) {
                TableColumn tableColumn = columns1.get(i);
                if (slaveSet.contains(tableColumn.getColumnName())){
                    continue;
                }
                colIndex.add(new int[]{1,i});
                TableColumn clone = tableColumn.clone();
                columns.add(clone);
            }
        //上下合并
        } else {
            for (MergeColumnDTO mergeColumnDTO : mergeColumnDTOS) {
                String master = mergeColumnDTO.getMaster();
                String slave = mergeColumnDTO.getSlave();
                String result = mergeColumnDTO.getResult();
                TableColumn masterCol = masterMap.get(master);
                TableColumn slaveCol = slaveMap.get(slave);
                if (!masterMap.containsKey(master) ) {
                    throw new ComponentException(String.format("主表列%s不存在", master));
                }
                if (StrUtil.isNotBlank(slave) && !"空".equals(slave)){
                    if (slaveMap.containsKey(slave)){
                        //当副表列为空跳过
                        if (!typeToBasicTypeInfo(masterCol.getDataType()).equals(typeToBasicTypeInfo(slaveCol.getDataType()))){
                            throw new ComponentException(String.format("主表列%s与副表列%s类型不匹配", master,slave));
                        }
                    } else {
                        throw new ComponentException(String.format("副表列%s不存在", slave));
                    }
                }
                if (masterMap.containsKey(master)){
                    TableColumn column = masterMap.get(master);
                    TableColumn clone = column.clone();
                    clone.setColumnName(result);
                    columns.add(clone);
                } else {
                    throw new ComponentException(String.format("主表列%s不存在", master));
                }
            }
        }
        return columns;
    }

    /**
     * 获取序号并设置序号
     * @param columnDTOS
     * @param mcolumns 主列
     * @param scolumns 副列
     */
    public void setIndex(List<MergeColumnDTO> columnDTOS,
                              List<TableColumn> mcolumns, List<TableColumn> scolumns, String mergeType) {
        List<String> mcolnames = mcolumns.stream().map(e -> e.getColumnName()).collect(Collectors.toList());
        List<String> scolnames = scolumns.stream().map(e -> e.getColumnName()).collect(Collectors.toList());

        for (MergeColumnDTO columnDTO : columnDTOS) {
            int i = mcolnames.indexOf(columnDTO.getMaster());
            if (i == -1){
                throw new ComponentException("数据来源列和主表列不对应，数据来源列中缺少列" + columnDTO.getMaster());
            }
            columnDTO.setMasterIndex(i);
            // mergeType 如果是上下合并就可以是空 匹配不上也当空
            int j = scolnames.indexOf(columnDTO.getSlave());
            if (j == -1){
                if (MergeColumnDTO.JOIN.equalsIgnoreCase(mergeType)) {
                    throw new ComponentException("数据来源列和副表列不对应，数据来源列中缺少列" + columnDTO.getSlave());
                }
            }
            columnDTO.setSlaveIndex(j);
        }
    }


    @Override
    public String getRule() {
        return ComponentCodeConstants.GATHER_MERGE;
    }
}
